#include "include/event_manager.h"

#include "rpc/net/socket.h"
#include "rpc/net/buffer.h"
#include "rpc/rpc_util.h"
#include "rpc/net/socket_opt.h"
#include "rpc/net/connection.h"
#include "rpc/net/proto_attr.h"

namespace {
void HandleTransEvent(int32 /*fd*/, void* ctx, int32 event);
}

namespace rpc {
Connection::Connection(EventManager* ev_mgr, Socket* socket)
    : ev_mgr_(ev_mgr), socket_(socket), protocol_(NULL) {
  CHECK_NOTNULL(socket_.get());
  input_buf_.reset(new Buffer);
  CHECK_NOTNULL(input_buf_.get());
  ev_.reset(new Event);
  CHECK_NOTNULL(ev_.get());
}

Connection::~Connection() {
}

bool Connection::Init() {
  ev_->fd = socket_->Fd();
  ev_->cb = HandleTransEvent;
  ev_->event = EV_READ;
  ev_->arg = this;

  return ev_mgr_->Add(ev_.get(), EV_READ);
}

void Connection::HandleRead() {
  if (!protocol_->HandleRead(this, input_buf_.get())) {
    return;
  }

  input_buf_->Revert();
  attr_->ReInit();
}

void Connection::HandleWrite() {
  sock::Send(socket_.get(), output_buf_.get());
  if (output_buf_->ReadableSize() == 0) {
    ev_mgr_->Mod(ev_.get(), EV_READ);
    output_buf_->Revert();
  }
}

void Connection::HandleError() {
  int error = 0;
  socklen_t errlen = sizeof(error);
  if (::getsockopt(socket_->Fd(), SOL_SOCKET, SO_ERROR, (void *) &error,
                   &errlen) == 0) {
    LOG(WARNING)<< "fd: " << socket_->Fd() << " error: " << strerror(error);
  }

  HandleClose();
}

void Connection::HandleClose() {
  AutoRunner r(close_closure_.get());
}

void Connection::ShutDown() {  //todo
}

int32 Connection::Recv(uint32 size) {
  if (size == 0) return 0;
  int32 readn = input_buf_->ReadFd(socket_->Fd(), size);
  if (readn == 0) {
    HandleClose();
    return 0;
  }

  return readn;
}

bool Connection::Send(const u_char* buf, uint32 buf_len) {
  if (buf_len == 0) return true;
  if (output_buf_->ReadableSize() != 0) {
    output_buf_->Append(buf, buf_len);
    return false;
  }

  int32 ret = sock::Send(socket_.get(), buf, buf_len);
  if (ret > 0) {
    if ((uint32) ret == buf_len) {
      return true;
    }
    buf += ret;
    buf_len -= ret;
  }

  output_buf_->Append(buf, buf_len);
  CHECK(ev_mgr_->Mod(ev_.get(), EV_READ | EV_WRITE));
  return false;
}

}  // end for namespace.

namespace {
void HandleTransEvent(int32 /*fd*/, void* ctx, int32 event) {
  rpc::Connection* trans = (rpc::Connection*) ctx;

  if (event & EV_READ) {
    trans->HandleRead();
  } else if (event & EV_WRITE) {
    trans->HandleWrite();
  } else {
    trans->HandleError();
  }
}
}
